package cluster

import (
	"errors"
	"gitee.com/cooge/impush/cluster/client"
	"gitee.com/cooge/impush/config"
	"gitee.com/cooge/impush/core"
	"gitee.com/cooge/impush/gutil"
	"gitee.com/cooge/impush/message"
	"gitee.com/cooge/impush/util"
	"github.com/sirupsen/logrus"
	"runtime"
	"strconv"
	"sync"
	"time"
)

var clientMap = gutil.NewLinkExMap()

func Start() {

	go startServer(config.CLUSTER_LOCAL_PORT)
	go startClient(config.CLUSTER_REMOTE_HOST, config.CLUSTER_REMOTE_PORT)
	go Live()
	initUserFilter()

}

func Live() {
	timer1 := time.NewTicker(time.Duration(30) * time.Second)
	for {
		select {
		case <-timer1.C:
			{

				syncMachine()

			}
		}
	}
}

var clientLock sync.Mutex

func startClient(rhost string, rport int) {
	clientLock.Lock()
	defer clientLock.Unlock()

	if len(rhost) > 0 && rport > 0 {
		addr := util.Merge(rhost, ":", strconv.Itoa(rport))
		if !clientMap.IsHasKey(addr) {
			logrus.Debug("连接新的机器",addr)
			cn := client.NewClusterClient2(addr)
			cn.Start(clusterClientHandle)
		}
	}
}

func clusterClientHandle(cc *client.ClusterClient, msg *message.Message, code int) error {

	switch code {

	case client.CLUSTER_CLIENT_START_SUCCESS:
		{
			clientMap.Add(cc.Raddress.String(), cc)

			go syncALLUser(cc)

			go syncOneMachine(cc)

		}
	case client.CLUSTER_CLIENT_START_ERROR:
		{
			clientMap.Delete(cc.Raddress.String(), cc)
		}
	case client.CLUSTER_CLIENT_NET_BREAK:
		{
			clientMap.Delete(cc.Raddress.String(), cc)
		}
	case client.CLUSTER_CLIENT_SEND_MESSAGE:
		{
			tu := msg.Heads.Get(message.TO)
			cs := core.GetUser(tu)
			if cs != nil {
				for _, u := range cs.GetAllArray() {
					if u.IsLocal() {
						_, err := core.SendMessage(tu, msg)
						return err
					}
				}
			}
			return errors.New(message.NO_USER)
		}
	case client.CLUSTER_CLIENT_SEND_USER:
		{
			tu := msg.Heads.Get(message.TO)
			sendAddUser(tu)
		}

	}
	core.AddInfo(config.CLUSTER_CONN_SERVER, clientMap.GetALLKey())
	return nil
}

func syncOneMachine(vv *client.ClusterClient) {
	keys := clientMap.GetALLKey()
	vv.Live(keys)
}

func sendDeleteUser(userId string,ti *time.Time) {
	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		var data = DeleteClusterUsers(userId,ti)

		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok {
					_, err := vv.Write(data)
					if err == nil {
						break
					}
				}
			}
		}
	}
}

func sendAddUser(userId ...string) {
	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		var data = SendClusterUsers(userId)
		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok {
					_, err := vv.Write(data)
					if err == nil {
						break
					}
				}
			}
		}
	}
}

func syncALLUser(clusterClient *client.ClusterClient) {
	us := core.GetALLUsers()
	urry := make([]string, 0)
	for index, k := range us {
		u := core.GetUser(k)
		if u != nil {
			if u.HasLocal() {
				urry = append(urry, k)
			}
		}
		if (index+1)>>7<<7 == (index+1) && len(urry) > 0 {
			data := SendClusterUsers(urry)
			_, err := clusterClient.Write(data)
			if err != nil {
				return
			}
			runtime.Gosched()
			urry = make([]string, 0)
		}
	}
	if len(urry) > 0 {
		data := SendClusterUsers(urry)
		_, err := clusterClient.Write(data)
		if err != nil {
			return
		}
	}
}

func syncMachine() {
	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok {
					_, err := vv.Live(keys)
					if err == nil {
						break
					}
				}
			}
		}
	}
}

func startServer(lport int) {
	ncs := NewClusterServer()
	err := ncs.Bind(lport)
	if err == nil {
		for {
			sc, err := ncs.Accept()
			if err == nil {
				sc.Start()
			} else {
				logrus.Panic("集群服务出现系统故障", err.Error())
				break
			}
		}
	} else {
		logrus.Panic("集群服务启动失败", err.Error())
	}
}
